package com.lantaiyuan.ebus.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;

import com.lantaiyuan.ebus.task.PushTask;

/***
 * 
* <p>Title: KafkaConsumer</p>
* <p>Description: 消费kafka异常数据</p>
* <p>Company: lty</p>
* @author liuhao
* @date 2017年5月5日 下午2:17:41
 */
public class KafkaConsumer {
	private static final Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);
	
	@Autowired
    private PushTask pushService;
	
	@KafkaListener(topics="${kafka.topic}")
	public void listen(ConsumerRecord<Integer, String> msg) {
		String jsonString = msg.value();
		logger.info("消费kafka数据：" + jsonString);
		
		//异步调用业务方法
		pushService.pushMsg(jsonString);
		logger.info("异步调用业务方法完毕");
	}
	
}